package com.digiwin.demo.job;

import com.digiwin.demo.constant.ConfigConstant;
import com.digiwin.demo.test.MyElasticsearchSinkFunction;
import com.digiwin.demo.test.MyMongoSinkFunction;
import com.digiwin.demo.util.RMQUtil;
import com.digiwin.demo.vo.VisitVO;
import com.google.gson.Gson;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Author yanggld
 * @Date 2019/12/26-14:16
 * @Description
 */
public class MqStreamJob {
	private static final Logger LOG = LoggerFactory.getLogger(MqStreamJob.class);
	public static void main(String[] args) {
		//1.获取flink的运行环境
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		RMQUtil<String> rmqUtil = new RMQUtil<>("*.book", ConfigConstant.RMQ_QUEUE_NAME);
		try {
			// 从消息队列获取流数据
			DataStream<String> dataStreamSource = rmqUtil.getDataStream(env);
			//输出Source的信息
			dataStreamSource.print();
			/**
			 * 数据转换
			 * MapFunction:第一个参数是你接收的数据的类型,第二个参数是返回数据的类型
			 */
			DataStream<VisitVO> visitVODataStream = dataStreamSource.map((MapFunction<String, VisitVO>) data -> {
				Gson gson = new Gson();
				if (StringUtils.isNotEmpty(data)) {
					return gson.fromJson(data, VisitVO.class);
				}
				return null;
			});
			// 传入ElasticsearchSinkFunction实现es对数据流的处理
			visitVODataStream.addSink(new MyElasticsearchSinkFunction().getElasticsearchSink());
			// 传入RichSinkFunction实现mongo对数据流的处理
			visitVODataStream.addSink(new MyMongoSinkFunction("test","book"));
			env.execute(Thread.currentThread().getStackTrace()[1].getClassName());
		}catch (Exception ex){
			LOG.error("MqStreamJob error*********************:"+ex.getMessage());
		}
	}
}
